package com.cloud.log.autolevel.manager;


import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @author: ZhangKaiQiang
 * @date: 2021/1/5
 * @time: 10:00
 *
 * zookeeper工具类
 **/
public class ZkManager {


    /**
     * 链接Url
     */
    private String connectionUrl;

    /**
     * session超时时间
     */
    private Integer sessionTimeOutMs;

    /**
     * 链接超时时间
     */
    private Integer connectionTimeoutMs;

    /**
     * 默认的zk路径
     */
    private String namespace;

    /**
     * 默认的session超时时间
     */
    private static final Integer DEFAULT_SESSION_TIMEOUT=30000;

    /**
     * 默认的connection超时时间
     */
    private static final Integer DEFAULT_CONNECTION_TIMEOUT=30000;

    private volatile CuratorFramework curatorFramework;

    private Logger logger= LoggerFactory.getLogger(ZkManager.class);

    public ZkManager(
            String connectionUrl,
            Integer sessionTimeOutMs,
            Integer connectionTimeoutMs,
            String namespace){
        this.connectionUrl=connectionUrl;
        this.sessionTimeOutMs=sessionTimeOutMs;
        this.connectionTimeoutMs=connectionTimeoutMs;
        this.namespace=namespace;
        this.init();
    }

    /**
     * 创建节点
     * @param path 节点路径
     * @param createMode 节点类型 永久还是临时
     * @param data 节点数据
     * @return 节点路径
     */
    public String createNode(String path,CreateMode createMode,byte[] data){
        if (StringUtils.isBlank(path) || createMode==null){
            return "";
        }

        try {
            curatorFramework.create().
                    creatingParentsIfNeeded().
                    withMode(createMode).
                    forPath(path, data);
        } catch (Exception e) {
            if (logger.isErrorEnabled()){
                logger.error(path+"创建节点失败",e);
            }
        }
        return path;
    }


    /**
     * 判断某个节点是否存在
     * @param path zk path
     * @return
     */
    public boolean checkIfExists(String path){
        if (StringUtils.isBlank(path)){
            return false;
        }

        try {
            Stat stat = curatorFramework.checkExists().forPath(path);
            if (stat==null){
                return false;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }


    /**
     * 获取某个节点的数据
     * @param path zk Path
     * @return byte[]
     */
    public byte[] getNodeData(String path){
        byte[] emptyByte=new byte[1];
        if (StringUtils.isBlank(path)){
            return emptyByte;
        }

        try {
            byte[] bytes = curatorFramework.getData().forPath(path);
            if (bytes!=null){
                return bytes;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return emptyByte;
    }


    /**
     * 删除某个节点
     * @param path zk Path
     */
    public void deleteNode(String path){
        if (StringUtils.isBlank(path)){
            return;
        }

        // 判断路径是否存在
        if (!checkIfExists(path)){
           return;
        }

        try {
            curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 获得某个节点的子节点
     * @param path 节点路径
     * @return List<String>
     */
    public List<String> getChildren(String path){
        if (StringUtils.isBlank(path)){
            return Collections.emptyList();
        }

       // 判断是否存在这个节点
        if (!checkIfExists(path)){
            return Collections.emptyList();
        }

        try {
            List<String> childrenPaths = curatorFramework.getChildren().forPath(path);
            return childrenPaths==null?Collections.emptyList():childrenPaths;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }


    /**
     * 更新某个节点的数据
     * @param path zk Path
     * @param data data
     * @return boolean
     */
    public boolean updateNodeData(String path,byte[] data){
        if (StringUtils.isBlank(path) || data==null){
            return false;
        }


        // 判断节点是否存在
        boolean checkResult = checkIfExists(path);
        if (!checkResult){
            return false;
        }

        try {
            Stat stat = curatorFramework.setData().forPath(path, data);
            if (stat!=null){
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * NodeCache用来观察ZNode自身,如果ZNode节点创建、更新(不包括删除),那么Node Cache会更新缓存，并触发事件给注册的监听器
     * @param path zk Path
     * @param consumer 对应的操作
     */
    public void createNodeCacheWatch(String path,Consumer<? super ChildData> consumer){
        if (StringUtils.isBlank(path)){
            if (logger.isWarnEnabled()){
                logger.warn("path is blank when create node cache watch");
            }
            return;
        }

        boolean checkResult = checkIfExists(path);
        if (!checkResult){
            if (logger.isWarnEnabled()){
                logger.warn("path is not present when create node cache watch");
            }
            return;
        }

        CuratorCache nodeCache = CuratorCache.build(curatorFramework, path, CuratorCache.Options.SINGLE_NODE_CACHE);
        CuratorCacheListener listener = CuratorCacheListener
                .builder()
                .forNodeCache(new NodeCacheListener() {
                    @Override
                    public void nodeChanged() throws Exception {
                        Optional<ChildData> childData = nodeCache.get(path);
                        childData.ifPresent(consumer);
                    }
                })
                .build();
        nodeCache.listenable().addListener(listener);
        nodeCache.start();
    }


    /**
     * 客户端连接初始化
     */
    private void init(){
        // 双重校验
        if (curatorFramework==null){
            synchronized (ZkManager.class){
                if (curatorFramework==null){

                    this.curatorFramework = CuratorFrameworkFactory.builder().
                            connectString(connectionUrl).
                            sessionTimeoutMs(sessionTimeOutMs==null?DEFAULT_SESSION_TIMEOUT: sessionTimeOutMs).
                            connectionTimeoutMs(connectionTimeoutMs==null?DEFAULT_CONNECTION_TIMEOUT: connectionTimeoutMs)
                            .canBeReadOnly(false)
                            .retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))
                            .build();
                    curatorFramework.start();
                    try {
                        // 阻塞判断连接成功，超时认为失败。
                        curatorFramework.blockUntilConnected(5, TimeUnit.SECONDS);
                        byte[] bytes = curatorFramework.getData().forPath("/");
                        if (curatorFramework.getZookeeperClient().isConnected()) {
                            // 返回连接，起始根目录为namespace。
                            String tempNamespace=namespace.replaceFirst("/","");
                            curatorFramework.usingNamespace(tempNamespace);
                            return;
                        }
                    } catch (InterruptedException ignored) {
                        Thread.currentThread().interrupt();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                    // 建立连接失败
                    curatorFramework.close();
                    throw new RuntimeException("failed to connect to zookeeper service : " + connectionUrl);
                }
            }
        }
    }

    public String getConnectionUrl() {
        return connectionUrl;
    }

    public Integer getSessionTimeOutMs() {
        return sessionTimeOutMs;
    }

    public Integer getConnectionTimeoutMs() {
        return connectionTimeoutMs;
    }

    public String getNamespace() {
        return namespace;
    }
}
